package com.wujian.love.service.impl;

import com.wujian.love.contants.RocketMQContants;
import com.wujian.love.domin.RocketMessage;
import com.wujian.love.mapper.OrderParentMapper;
import com.wujian.love.mapper.RocketMessageMapper;
import com.wujian.love.utils.JsonSerilizable;
import org.apache.rocketmq.client.producer.LocalTransactionState;
import org.apache.rocketmq.client.producer.TransactionListener;
import org.apache.rocketmq.common.message.Message;
import org.apache.rocketmq.common.message.MessageExt;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

import java.text.SimpleDateFormat;
import java.util.Date;
import java.util.Map;

/**
 * @ClassName: TransactionListenerImp1
 * @description:
 * @author: wujian
 * @Date: 2020-03-21 23:27
 **/
@Component
public class TransactionListenerImpl implements TransactionListener {

    private static volatile SimpleDateFormat sdf = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");

    @Autowired
    private OrderParentMapper orderParentMapper;

    @Autowired
    private RocketMessageMapper rocketMessageMapper;


    @Override
    public LocalTransactionState executeLocalTransaction(Message msg, Object arg) {
        try {
            System.out.println("消息监听:当前时间" + sdf.format(new Date()));

            //事务消息
            String msgType = (String) arg;
            System.out.println("消息类型" + msgType);
            if (msgType.equals(RocketMQContants.STOCK_ORDER_MSG_TYPE)) {
                Map<String, Object> paramsBody = JsonSerilizable.deserilizableForMapFromFile(new String(msg.getBody()), Object.class);
                System.out.println("消息监听:扣减库存" + sdf.format(new Date()));
                String keys = msg.getKeys();
                //将消息状态改为已消费
                int row = rocketMessageMapper.updateRocketMessageByKeys(keys);
            }

            return LocalTransactionState.COMMIT_MESSAGE;
        } catch (Exception e) {
            e.printStackTrace();
        }
        return LocalTransactionState.ROLLBACK_MESSAGE;
    }

    @Override
    public LocalTransactionState checkLocalTransaction(MessageExt msg) {
        System.out.println("====================================回调消息检查=========================");
        //检查消息是否已经消费
        RocketMessage rocketMessage = rocketMessageMapper.selectRocketMessageByKeys(msg.getKeys());
        if (rocketMessage.getConsumeFlag().equals(RocketMessage.CONSUME_FlAG_TRUE)) {
            return LocalTransactionState.COMMIT_MESSAGE;
        } else {
            //消息未被消费
            return LocalTransactionState.ROLLBACK_MESSAGE;
        }

    }
}
